package cn.itcast.czxy.BD18

import java.util.Properties

import cn.itcast.czxy.BD18.bean.{TagesRule, Tagsfour}
import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object SexTag {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().master("local[*]").appName("sextag").getOrCreate()
    var url="jdbc:mysql://bd001:3306/tags_new?userUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&user=root&password=123456"
    var tableName="tbl_basic_tag"
    var proerties:Properties=new Properties
    val mysqlconn: DataFrame = spark.read.jdbc(url,tableName,proerties).persist()
    val sijitag: Dataset[Row] = mysqlconn.select("id","rule").where("id=61")
    import spark.implicits._
    import scala.collection.JavaConverters._

    import org.apache.spark.sql.functions._


    val kvmap: Map[String, String] = sijitag.map(row => {
      row.getAs("rule").toString.split("##").map(kv => {
        val strings: Array[String] = kv.split("=")
        (strings(0), strings(1))
      })

    }).collectAsList().get(0).toMap
    val tagsfour = toHbaseMeta(kvmap)
//    println(tagsfour)

    val fiveTage: Dataset[Row] = mysqlconn.select("id","rule").where("pid=61")
    val fiveTageList: List[TagesRule] = fiveTage.map(row => {
      val id: Int = row.getAs("id").toString.toInt
      val rule: String = row.getAs("rule").toString
      TagesRule(id, rule)
    }).collectAsList().asScala.toList


//    fiveTageList.foreach(println)

    val hbaseDatas = spark.read.format("cn.itcast.czxy.BD18.cn.itcast.czxy.BD18.tools.HBaseDataSource")
      .option("zkHosts", tagsfour.zkHosts)
      .option("zkPort", tagsfour.zkPort)
      .option("hbaseTable", tagsfour.hbaseTable)
      .option("family", tagsfour.family)
      .option("selectFields", tagsfour.selectFields)
      .load()

    //udf,性别替换标签
    var  GetTagId=udf((rule:String)=>{
      var id=0
      for (elem <- fiveTageList) {
          if (rule==elem.rule){
            id=elem.id
          }
      }
      id
    })

    val userTags: DataFrame = hbaseDatas.select('id as ("userId"),GetTagId('gender).as("tagsId"))

    userTags.write.format("cn.itcast.czxy.BD18.cn.itcast.czxy.BD18.tools.HBaseDataSource")
      .option("zkHosts", tagsfour.zkHosts)
      .option("zkPort", tagsfour.zkPort)
      //存储的表
      .option("hbaseTable", "test")
      .option("family", "detail")
      .option("selectFields", "userId,tagsId")
      .save()
  }





  def toHbaseMeta(kvmap:Map[String,String]):Tagsfour={
    Tagsfour(kvmap.getOrElse(Tagsfour.INTYPE,""),
      kvmap.getOrElse("zkHosts",""),
      kvmap.getOrElse("zkPort",""),
      kvmap.getOrElse("hbaseTable",""),
      kvmap.getOrElse("family",""),
      kvmap.getOrElse("selectFields",""),
      kvmap.getOrElse("rowkey",""))
  }
}
